package com.alibaba.otter.node.etl.load.loader.mq.rabbit.consume;

import com.alibaba.otter.node.etl.load.loader.mq.rabbit.MqUtils;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.callback.SuccessCallback;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.exception.IllegalMessageException;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.exception.MqBusinessException;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.exception.MqSerializeException;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.model.MessageContext;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.provider.DefaultSerializerProvider;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.provider.DefaultTargetClassExplorer;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.provider.SerializerProvider;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.provider.TargetClassExplorer;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.serializer.SerializeUtils;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.rabbitmq.client.AMQP;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;

import java.lang.reflect.Field;
import java.util.List;
import java.util.concurrent.ExecutionException;

@Slf4j
@SuppressWarnings({"rawtypes", "unchecked"})
public class DefaultHandlerInvoker implements HandlerInvoker {

    @Getter
    private Handler handler;

    @Setter
    private List<PropertyHandler> propertyHandlers;

    @Setter
    private List<HeaderHandler> headerHandlers;

    @Setter
    private List<SuccessCallback> successCallbacks;

    public void invoke(MessageContext context) {

        boolean isAdvancedHandler = handler instanceof AdvancedHandler;

        // handle header and properties, if fail, log and continue.
        AMQP.BasicProperties properties = (AMQP.BasicProperties) context.getMessage().getProperty();

        if (!isAdvancedHandler || ((AdvancedHandler) handler).useDefaultHeaderHandler(context)) {
            headerHandlers.forEach(handler -> handleHeader(handler, properties));
        }
        if (!isAdvancedHandler || ((AdvancedHandler) handler).useDefaultPropertiesHandler(context)) {
            propertyHandlers.forEach(handler -> handleProperty(handler, properties));
        }

        // check if message are rabbitmq message.
        if (!isAdvancedHandler || ((AdvancedHandler) handler).checkMessageType(context)) {
            checkMessageType(context);
        }

        if (!isAdvancedHandler || ((AdvancedHandler) handler).useDefaultSerializer(context)) {
            // set deserialized message to message context for log use.
            context.setDeserializedMsg(deserializeMessage(context));
        }

        // invoke user's handler.
        doInvoke(context, properties);

        // ack message.
        if (!isAdvancedHandler || ((AdvancedHandler) handler).useDefaultAckStrategy(context)) {
            MqUtils.ack(context);
        }

        invokeSuccessCallbacks(context);
    }

    private void invokeSuccessCallbacks(MessageContext ctx) {
        if (CollectionUtils.isEmpty(successCallbacks)) {
            return;
        }
        successCallbacks.forEach(callback -> {
            try {
                callback.onSuccess(ctx.getDeserializedMsg());
            } catch (Exception e) {
                log.error("exception occur on invoking success callback.", e);
            }
        });
    }

    private void checkMessageType(MessageContext context) {
        if (!(context.getMessage().getProperty() instanceof AMQP.BasicProperties)) {
            throw new IllegalMessageException("Not a rabbitmq message.", context);
        }
    }

    private void doInvoke(MessageContext context, AMQP.BasicProperties properties) {
        try {
            if (handler instanceof AdvancedHandler) {
                handler.handle(context);
            } else {
                handler.handle(context.getDeserializedMsg());
            }
        } catch (ClassCastException e) {
            String errMsg = "消息类型与接口入参不匹配";
            log.error(errMsg, e);
            Class clz = context.getDeserializedMsg() == null ? null : context.getDeserializedMsg().getClass();
            throw new MqSerializeException(errMsg, clz, serializerProvider.provide(properties, clz), e);
        } catch (Exception e) {
            String errMsg = "消息消费出错";
            log.error(errMsg, e);
            throw new MqBusinessException(errMsg, context, e);
        }
    }

    private SerializerProvider serializerProvider = new DefaultSerializerProvider();
    private TargetClassExplorer classExplorer;

    private Object deserializeMessage(MessageContext context) {
        return SerializeUtils.deserializeMessage(
                context,
                serializerProvider,
                classExplorer
        );
    }

    public void setHandler(Handler handler) {
        this.handler = handler;
        this.classExplorer = new DefaultTargetClassExplorer(handler);
    }

    private void handleHeader(HeaderHandler handler, AMQP.BasicProperties properties) {
        try {
            handler.handleHeader(properties.getHeaders());
        } catch (Exception e) {
            String errMsg = "HeaderHandler[" + handler.getClass().getName() + "]执行失败";
            log.error(errMsg, e);
        }
    }

    private void handleProperty(PropertyHandler handler, AMQP.BasicProperties properties) {
        try {
            Object fieldValue = fieldCache.get(handler.propertyName()).get(properties);
            handler.handleProperty(fieldValue);
        } catch (IllegalAccessException | ExecutionException e) {
            String errMsg = "PropertyHandler[" + handler.getClass().getName() + "]执行失败";
            log.error(errMsg, e);
        }
    }

    private LoadingCache<String, Field> fieldCache = CacheBuilder.newBuilder().
            build(new CacheLoader<String, Field>() {
                @Override
                public Field load(String propertyName) throws Exception {
                    Field field = AMQP.BasicProperties.class.getDeclaredField(propertyName);
                    field.setAccessible(true);
                    return field;
                }
            });
}
